package InputFormat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;


public class FilesRecordReader extends RecordReader <Text, BytesWritable> {
    private Text k = new Text();
    private BytesWritable v = new BytesWritable();
    private FileSplit split;
    private Configuration configuration;
    boolean isProgress=true;


    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
      //初始化，切片内容，配置文件
        this.split=(FileSplit) inputSplit;
        configuration = taskAttemptContext.getConfiguration();



    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        //主要逻辑
        if (isProgress) {

            //获取文件系统对象
            Path path = split.getPath();
            FileSystem fs = path.getFileSystem(configuration);
            //获取输入流
            FSDataInputStream fis = fs.open(path);

            //拷贝
            byte[] buffer = new byte[(int) split.getLength()];
            IOUtils.readFully(fis, buffer, 0, buffer.length);

            //封装数据
            k.set(path.toString());
            v.set(buffer, 0, buffer.length);
            //关流
            IOUtils.closeStream(fis);

            //代表这一个文件已读完
            isProgress=false;
            //通过map端写入map里（this.map(context.getCurrentKey(), context.getCurrentValue(), context);）
            return  true;


        }
        //保证不会读同一个数据（this.cleanup(context);）
        return false;
    }
    //表示当前的状态
    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
       //返回对应封装好的数据
        return k;
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {

        return v;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {
    }
}
